package com.yeyks.rocketmq;

import cn.hutool.json.JSONUtil;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.yeyks.common.dto.mq.EsUpdateMessage;
import com.yeyks.search.constants.EsIndexConstant;
import com.yeyks.search.service.hotel.EsImportHotelService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

/**
 * @program: cnbooking
 * @description:
 * @author: chenzheng
 * @create: 2019-10-12 14:36
 **/

@Slf4j
@Component
public class MessageESUpdateListener implements MessageListener {

    @Autowired
    private EsImportHotelService esImportHotelService;

    @Override
    public Action consume(Message message, ConsumeContext consumeContext) {
        try {
            Long startTime = System.currentTimeMillis();
            byte[] body = message.getBody();
            //获取到接收的消息，由于接收到的是byte数组，所以需要转换成字符串
            String msg = new String(body);
            //处理业务逻辑
            disposeLogic(msg);
            log.info("esupdate更新检测：监听到的消息：message = {}", msg);
            Long endTime = System.currentTimeMillis();
            log.info("esupdate检测：单次消费耗时：" + (endTime - startTime) / 1000);
        } catch (Exception e) {
            log.error("MessageListener.consume error:" + e.getMessage(), e);
        }
        return Action.CommitMessage;
    }

    /**
     * 处理监听到的房态数据更新，更具类型同步更新es中酒店房态数据
    * @param msg
     */
    private void disposeLogic(String msg) {
        try {
            EsUpdateMessage esUpdateMessage = JSONUtil.toBean(msg, EsUpdateMessage.class);
            if (null != esUpdateMessage) {

                if(EsIndexConstant.JOIN_TYPE_HOTEL.equals(esUpdateMessage.getType())){
                    if("cn".equals(esUpdateMessage.getSourceType())){
                        if(!CollectionUtils.isEmpty(esUpdateMessage.getSourceIds())){
                            esImportHotelService.importCnBookingHotelByHotelId(Integer.valueOf(esUpdateMessage.getSourceIds().get(0)),
                                    EsIndexConstant.INDEX_HOTEL2, "0001","","");
                        }
                    }else{
                        esImportHotelService.importHotelByHotelIdList(esUpdateMessage.getSourceIds(), EsIndexConstant.INDEX_HOTEL2);
                    }
                }else if(EsIndexConstant.JOIN_TYPE_ROOM.equals(esUpdateMessage.getType())){
                    esImportHotelService.importRoomByRoomIdList(esUpdateMessage.getSourceIds(),EsIndexConstant.INDEX_HOTEL2);
                }else if(EsIndexConstant.JOIN_TYPE_CALENDAR.equals(esUpdateMessage.getType())){
                    esImportHotelService.importCalendarByCalendarIdList(esUpdateMessage.getSourceIds(),EsIndexConstant.INDEX_HOTEL2);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("webSocket检测订单退款：处理监听消息异常：msg = {}", msg);
        }
    }
}
